package com.oromwin.repos.server.thread;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.logging.Log;

import com.oromwin.repos.common.Common;
import com.oromwin.repos.common.Logger;
import com.oromwin.repos.common.ReposException;
import com.oromwin.repos.common.Common.LockType;
import com.oromwin.repos.common.helper.IOHelper;
import com.oromwin.repos.server.LockObject;
import com.oromwin.repos.socket.request.Request;
import com.oromwin.repos.socket.response.Response;

/*
 * XXX: how about splitting input output thread, need to share client socket 
 */
public class SocketThread extends Thread {
    private Log log = Logger.getLog(SocketThread.class);
    private Socket _socket = null;
    private Map<String, LockObject> _lock = null;
    private LockObject _lockObject = null;
    
    public SocketThread(Socket socket) throws IOException {
        _socket = socket;
        _lock = new HashMap<String, LockObject>();
    }
    
    public void run() {
        Request req = null;
        Response res = null;
        String className = null;
        try {
            DataInputStream dis = new DataInputStream(new BufferedInputStream(_socket.getInputStream()));
            className = IOHelper.readUTF8(dis);
            req = (Request) Class.forName(className).newInstance();
            LockType lockType = Common.getLockTypeOfPacket(req);
            String id = req.id();
            synchronized (_lock) { // global lock
                _lockObject = _lock.get(id);
                if (_lockObject == null) {
                    _lockObject = new LockObject(lockType, id);
                    _lock.put(id, _lockObject);

                } else if (_lockObject.getLockType() == Common.LockType.READ && lockType == Common.LockType.READ) {
                    _lockObject = new LockObject(lockType, id);
                    _lock.put(id, _lockObject);

                } else {
                    log.debug("Read request came from client while writing or vice versa");
                }
            }
            synchronized (_lockObject) { // local lock
                try {
                    req.receive(dis, _lockObject);
                    res = req.run();
                } finally {
                    _lock.remove(id);
                }
            }
            DataOutputStream dos = new DataOutputStream(new BufferedOutputStream(_socket.getOutputStream()));
            res.send(dos);
            dos.flush();
            dos.close();
            dis.close();
            if (log.isDebugEnabled()) log.debug(String.format("Result, {%s}", res.getResult().toString()));
        } catch (ClassNotFoundException e) {
            String message = String.format("Illegal className, {%s}", className);
            Logger.errorWithThrowable(log, e, message);
        } catch (InstantiationException e) {
            String message = String.format("Illegal default constructor, {%s()}", className);
            Logger.errorWithThrowable(log, e, message);
        } catch (IllegalAccessException e) {
            String message = String.format("Private or protected constructor, {%s()}", className);
            Logger.errorWithThrowable(log, e, message);
        } catch (IOException e) {
            String message = String.format("Illegal socket stream, {%s}", _socket.toString());
            Logger.errorWithThrowable(log, e, message);
        } catch (ReposException e) {
            Logger.errorWithThrowable(log, e, null);
        }
    }

}
